#include "cloud_msg.h"
#include <stdio.h>
#include <unistd.h>
#include <string.h>
#include <pthread.h>
#include <stdlib.h>
#include <fcntl.h>
#include "aiot_state_api.h"
#include "aiot_sysdep_api.h"
#include "aiot_dynregmq_api.h"
#include "aiot_mqtt_api.h"
#include "aiot_subdev_api.h"
#include "aiot_ota_api.h"
#include "aiot_ra_api.h"
#include "queue.h"
#include "config.h"
#include "aliyun_proc.h"
#include "nlohmann/json.hpp"
using json = nlohmann::json;
/* 
    原端口：1883/443，对应的证书(GlobalSign R1),于2028年1月过期，届时可能会导致设备不能建连。
    (推荐)新端口：8883，将搭载新证书，由阿里云IoT自签证书，于2053年7月过期。
*/
const uint16_t port = 8883;
/* TODO: 如果要免预注册, 需要将该值设置为1;如果需要在控制台预先注册设备, 置为0 */
uint8_t skip_pre_regist = 0;

/* 位于portfiles/aiot_port文件夹下的系统适配函数集合 */
extern aiot_sysdep_portfile_t g_aiot_sysdep_portfile;

/* 位于external/ali_ca_cert.c中的服务器证书 */
extern const char *ali_ca_cert;

/* 白名单模式下用于保存deviceSecret的结构体定义 */
typedef struct {
	char device_secret[64];
} demo_devinfo_wl_t;
/* 用户保存白名单模式动态注册, 服务器返回的deviceSecret */
static demo_devinfo_wl_t demo_devinfo_wl;

/* 免白名单模式下用于保存mqtt建连信息clientid, username和password的结构体定义 */
typedef struct {
	char conn_clientid[128];
	char conn_username[128];
	char conn_password[64];
} demo_devinfo_nwl_t;
/* 用户保存免白名单模式动态注册, 服务器返回的mqtt建连信息 */
static demo_devinfo_nwl_t demo_devinfo_nwl;

/* 日志回调函数, SDK的日志会从这里输出 */
static int32_t demo_state_logcb(int32_t code, char *message)
{
	if (STATE_HTTP_LOG_RECV_CONTENT != code) {
		printf("%s", message);
	}
	return 0;
}

void demo_dynregmq_recv_handler(void *handle, const aiot_dynregmq_recv_t *packet, void *userdata)
{
	switch (packet->type) {
	/* TODO: 回调中需要将packet指向的空间内容复制保存好, 因为回调返回后, 这些空间就会被SDK释放 */
	case AIOT_DYNREGMQRECV_DEVICEINFO_WL: {
		if (strlen(packet->data.deviceinfo_wl.device_secret) >= sizeof(demo_devinfo_wl.device_secret)) {
			break;
		}

		/* 白名单模式, 用户务必要对device_secret进行持久化保存 */
		memset(&demo_devinfo_wl, 0, sizeof(demo_devinfo_wl_t));
		memcpy(demo_devinfo_wl.device_secret, packet->data.deviceinfo_wl.device_secret, strlen(packet->data.deviceinfo_wl.device_secret));
	} break;
	/* TODO: 回调中需要将packet指向的空间内容复制保存好, 因为回调返回后, 这些空间就会被SDK释放 */
	case AIOT_DYNREGMQRECV_DEVICEINFO_NWL: {
		if (strlen(packet->data.deviceinfo_nwl.clientid) >= sizeof(demo_devinfo_nwl.conn_clientid) ||
		    strlen(packet->data.deviceinfo_nwl.username) >= sizeof(demo_devinfo_nwl.conn_username) ||
		    strlen(packet->data.deviceinfo_nwl.password) >= sizeof(demo_devinfo_nwl.conn_password)) {
			break;
		}

		/* 免白名单模式, 用户务必要对MQTT的建连信息clientid, username和password进行持久化保存 */
		memset(&demo_devinfo_nwl, 0, sizeof(demo_devinfo_nwl_t));
		memcpy(demo_devinfo_nwl.conn_clientid, packet->data.deviceinfo_nwl.clientid, strlen(packet->data.deviceinfo_nwl.clientid));
		memcpy(demo_devinfo_nwl.conn_username, packet->data.deviceinfo_nwl.username, strlen(packet->data.deviceinfo_nwl.username));
		memcpy(demo_devinfo_nwl.conn_password, packet->data.deviceinfo_nwl.password, strlen(packet->data.deviceinfo_nwl.password));
	} break;
	default: {
	} break;
	}
}

int dyn_register(const char *pkey, const char *psec, const char *dname, const char *host, char **dsec)
{
	int32_t res = STATE_SUCCESS;
	void *dynregmq_handle = NULL;
	aiot_sysdep_network_cred_t cred; /* 安全凭据结构体, 如果要用TLS, 这个结构体中配置CA证书等参数 */

	/* 配置SDK的底层依赖 */
	aiot_sysdep_set_portfile(&g_aiot_sysdep_portfile);
	/* 配置SDK的日志输出 */
	aiot_state_set_logcb(demo_state_logcb);

	const char *product_key = pkey;
	const char *device_name = dname;
	const char *product_secret = psec;
	const char *mqtt_host = host;

	/* 创建SDK的安全凭据, 用于建立TLS连接 */
	memset(&cred, 0, sizeof(aiot_sysdep_network_cred_t));
	cred.option = AIOT_SYSDEP_NETWORK_CRED_SVRCERT_CA; /* 使用RSA证书校验DYNREGMQ服务端 */
	cred.max_tls_fragment = 16384; /* 最大的分片长度为16K, 其它可选值还有4K, 2K, 1K, 0.5K */
	cred.sni_enabled = 1; /* TLS建连时, 支持Server Name Indicator */
	cred.x509_server_cert = ali_ca_cert; /* 用来验证服务端的RSA根证书 */
	cred.x509_server_cert_len = strlen(ali_ca_cert); /* 用来验证服务端的RSA根证书长度 */

	/* 创建1个dynregmq客户端实例并内部初始化默认参数 */
	dynregmq_handle = aiot_dynregmq_init();
	if (dynregmq_handle == NULL) {
		printf("aiot_dynregmq_init failed\n");
		return -1;
	}

	/* 配置连接的服务器地址 */
	res = aiot_dynregmq_setopt(dynregmq_handle, AIOT_DYNREGMQOPT_HOST, (void *)mqtt_host);
	if (res < STATE_SUCCESS) {
		printf("aiot_dynregmq_setopt AIOT_DYNREGMQOPT_HOST failed, res: -0x%04X\n", -res);
		aiot_dynregmq_deinit(&dynregmq_handle);
		return -1;
	}

	/* 配置连接的服务器端口 */
	res = aiot_dynregmq_setopt(dynregmq_handle, AIOT_DYNREGMQOPT_PORT, (void *)&port);
	if (res < STATE_SUCCESS) {
		printf("aiot_dynregmq_setopt AIOT_DYNREGMQOPT_PORT failed, res: -0x%04X\n", -res);
		aiot_dynregmq_deinit(&dynregmq_handle);
		return -1;
	}

	/* 配置设备productKey */
	res = aiot_dynregmq_setopt(dynregmq_handle, AIOT_DYNREGMQOPT_PRODUCT_KEY, (void *)product_key);
	if (res < STATE_SUCCESS) {
		printf("aiot_dynregmq_setopt AIOT_DYNREGMQOPT_PRODUCT_KEY failed, res: -0x%04X\n", -res);
		aiot_dynregmq_deinit(&dynregmq_handle);
		return -1;
	}

	/* 配置设备productSecret */
	res = aiot_dynregmq_setopt(dynregmq_handle, AIOT_DYNREGMQOPT_PRODUCT_SECRET, (void *)product_secret);
	if (res < STATE_SUCCESS) {
		printf("aiot_dynregmq_setopt AIOT_DYNREGMQOPT_PRODUCT_SECRET failed, res: -0x%04X\n", -res);
		aiot_dynregmq_deinit(&dynregmq_handle);
		return -1;
	}

	/* 配置设备deviceName */
	res = aiot_dynregmq_setopt(dynregmq_handle, AIOT_DYNREGMQOPT_DEVICE_NAME, (void *)device_name);
	if (res < STATE_SUCCESS) {
		printf("aiot_dynregmq_setopt AIOT_DYNREGMQOPT_DEVICE_NAME failed, res: -0x%04X\n", -res);
		aiot_dynregmq_deinit(&dynregmq_handle);
		return -1;
	}

	/* 配置网络连接的安全凭据, 上面已经创建好了 */
	res = aiot_dynregmq_setopt(dynregmq_handle, AIOT_DYNREGMQOPT_NETWORK_CRED, (void *)&cred);
	if (res < STATE_SUCCESS) {
		printf("aiot_dynregmq_setopt AIOT_DYNREGMQOPT_NETWORK_CRED failed, res: -0x%04X\n", -res);
		aiot_dynregmq_deinit(&dynregmq_handle);
		return -1;
	}

	/* 配置DYNREGMQ默认消息接收回调函数 */
	res = aiot_dynregmq_setopt(dynregmq_handle, AIOT_DYNREGMQOPT_RECV_HANDLER, (void *)demo_dynregmq_recv_handler);
	if (res < STATE_SUCCESS) {
		printf("aiot_dynregmq_setopt AIOT_DYNREGMQOPT_RECV_HANDLER failed, res: -0x%04X\n", -res);
		aiot_dynregmq_deinit(&dynregmq_handle);
		return -1;
	}

	/* 配置DYNREGMQ动态注册模式,
        1. 配置为0则为白名单模式, 用户必须提前在控制台录入deviceName, 动态注册完成后服务会返回deviceSecret, 用户可通过
           AIOT_DYNREGMQRECV_DEVICEINFO_WL类型数据回调获取到deviceSecret.
        2. 配置为1则为免白名单模式, 用户无需提前在控制台录入deviceName, 动态注册完成后服务会返回MQTT建连信息, 用户可通过
           AIOT_DYNREGMQRECV_DEVICEINFO_NWL类型数据回调获取到clientid, username, password. 用户需要将这三个参数通过
           aiot_mqtt_setopt接口以AIOT_MQTTOPT_CLIENTID, AIOT_MQTTOPT_USERNAME, AIOT_MQTTOPT_PASSWORD配置选项
           配置到MQTT句柄中。
        */
	res = aiot_dynregmq_setopt(dynregmq_handle, AIOT_DYNREGMQOPT_NO_WHITELIST, (void *)&skip_pre_regist);
	if (res < STATE_SUCCESS) {
		printf("aiot_dynregmq_setopt AIOT_DYNREGMQOPT_NO_WHITELIST failed, res: -0x%04X\n", -res);
		aiot_dynregmq_deinit(&dynregmq_handle);
		return -1;
	}

	/* 发送动态注册请求 */
	res = aiot_dynregmq_send_request(dynregmq_handle);
	if (res < STATE_SUCCESS) {
		printf("aiot_dynregmq_send_request failed: -0x%04X\n\r\n", -res);
		printf("please check variables like mqtt_host, produt_key, device_name, product_secret in demo\r\n");
		aiot_dynregmq_deinit(&dynregmq_handle);
		return -1;
	}

	/* 接收动态注册请求 */
	res = aiot_dynregmq_recv(dynregmq_handle);
	if (res < STATE_SUCCESS) {
		printf("aiot_dynregmq_recv failed: -0x%04X\n", -res);
		aiot_dynregmq_deinit(&dynregmq_handle);
		return -1;
	}

	/* 把服务应答中的信息打印出来 */
	if (skip_pre_regist == 0) {
		printf("device secret: %s\n", demo_devinfo_wl.device_secret);

		/**
                预注册的动态注册的设备在获取到秘钥后,
                可以在mqtt_basic_demo.c中设置三元组(包括device_secret这个变量)，
                再调用aiot_mqtt_connect连云
            */
		/*

            */
	} else {
		printf("clientid: %s\n", demo_devinfo_nwl.conn_clientid);
		printf("username: %s\n", demo_devinfo_nwl.conn_username);
		printf("password: %s\n", demo_devinfo_nwl.conn_password);

		/**
                免预注册的动态注册的设备在获取到秘钥后,
                可以在mqtt_basic_demo.c中通过如下方式设置秘钥,
                再调用aiot_mqtt_connect连云
            */
		/*
                aiot_mqtt_setopt(mqtt_handle, AIOT_MQTTOPT_USERNAME, demo_devinfo_nwl.conn_username);
                aiot_mqtt_setopt(mqtt_handle, AIOT_MQTTOPT_PASSWORD, demo_devinfo_nwl.conn_password);
                aiot_mqtt_setopt(mqtt_handle, AIOT_MQTTOPT_CLIENTID, demo_devinfo_nwl.conn_clientid);
            */
	}

	/* 销毁动态注册会话实例 */
	res = aiot_dynregmq_deinit(&dynregmq_handle);
	if (res < STATE_SUCCESS) {
		printf("aiot_dynregmq_deinit failed: -0x%04X\n", -res);
		return -1;
	}
	*dsec = demo_devinfo_wl.device_secret;
	return 0;
}

/********************************/
static pthread_t g_mqtt_process_thread;
static pthread_t g_mqtt_recv_thread;
static uint8_t g_mqtt_process_thread_running = 0;
static uint8_t g_mqtt_recv_thread_running = 0;
static int g_aly_status = 0;

int aly_is_online()
{
	return g_aly_status;
}

/* MQTT事件回调函数, 当网络连接/重连/断开时被触发, 事件定义见core/aiot_mqtt_api.h */
void demo_mqtt_event_handler(void *handle, const aiot_mqtt_event_t *event, void *userdata)
{
	switch (event->type) {
	/* SDK因为用户调用了aiot_mqtt_connect()接口, 与mqtt服务器建立连接已成功 */
	case AIOT_MQTTEVT_CONNECT: {
		g_aly_status = 1;
		cloud_online();
		printf("AIOT_MQTTEVT_CONNECT\n");
		/* TODO: 处理SDK建连成功, 不可以在这里调用耗时较长的阻塞函数 */
	} break;

	/* SDK因为网络状况被动断连后, 自动发起重连已成功 */
	case AIOT_MQTTEVT_RECONNECT: {
		g_aly_status = 1;
		const char *recon = R"({"Active":"CLOUD_ONLINE_EVENT"})";
		send_msg(CTRL_QUEUE, recon, strlen(recon));
		printf("AIOT_MQTTEVT_RECONNECT\n");
		/* TODO: 处理SDK重连成功, 不可以在这里调用耗时较长的阻塞函数 */
	} break;

	/* SDK因为网络的状况而被动断开了连接, network是底层读写失败, heartbeat是没有按预期得到服务端心跳应答 */
	case AIOT_MQTTEVT_DISCONNECT: {
		g_aly_status = 0;
		const char *discon = R"({"Active":"CLOUD_OFFLINE_EVENT"})";
		send_msg(CTRL_QUEUE, discon, strlen(discon));

		const char *cause = (event->data.disconnect == AIOT_MQTTDISCONNEVT_NETWORK_DISCONNECT) ? ("network disconnect") : ("heartbeat disconnect");
		printf("AIOT_MQTTEVT_DISCONNECT: %s\n", cause);
		/* TODO: 处理SDK被动断连, 不可以在这里调用耗时较长的阻塞函数 */
	} break;

	default: {
	}
	}
}

/* MQTT默认消息处理回调, 当SDK从服务器收到MQTT消息时, 且无对应用户回调处理时被调用 */
void demo_mqtt_default_recv_handler(void *handle, const aiot_mqtt_recv_t *packet, void *userdata)
{
	switch (packet->type) {
	case AIOT_MQTTRECV_HEARTBEAT_RESPONSE: {
		printf("heartbeat response\n");
		/* TODO: 处理服务器对心跳的回应, 一般不处理 */
	} break;

	case AIOT_MQTTRECV_SUB_ACK: {
		printf("suback, res: -0x%04X, packet id: %d, max qos: %d\n", -packet->data.sub_ack.res, packet->data.sub_ack.packet_id, packet->data.sub_ack.max_qos);
		/* TODO: 处理服务器对订阅请求的回应, 一般不处理 */
	} break;

	case AIOT_MQTTRECV_PUB: {
		printf("pub, qos: %d, topic: %.*s\n", packet->data.pub.qos, packet->data.pub.topic_len, packet->data.pub.topic);
		printf("pub, payload: %.*s\n", packet->data.pub.payload_len, packet->data.pub.payload);

		nlohmann::json js;
		js["Topic"] = string(packet->data.pub.topic, packet->data.pub.topic_len);
		js["Payload"] = json::parse(packet->data.pub.payload, packet->data.pub.payload + packet->data.pub.payload_len);
		string tok = js.dump(4);
		send_msg(CTRL_QUEUE, tok.c_str(), tok.size());
		/* TODO: 处理服务器下发的业务报文 */
	} break;

	case AIOT_MQTTRECV_PUB_ACK: {
		printf("puback, packet id: %d\n", packet->data.pub_ack.packet_id);
		/* TODO: 处理服务器对QoS1上报消息的回应, 一般不处理 */
	} break;

	default: {
	}
	}
}

/* 执行aiot_mqtt_process的线程, 包含心跳发送和QoS1消息重发 */
void *demo_mqtt_process_thread(void *args)
{
	int32_t res = STATE_SUCCESS;

	while (g_mqtt_process_thread_running) {
		res = aiot_mqtt_process(args);
		if (res == STATE_USER_INPUT_EXEC_DISABLED) {
			break;
		}
		sleep(1);
	}
	return NULL;
}

/* 执行aiot_mqtt_recv的线程, 包含网络自动重连和从服务器收取MQTT消息 */
void *demo_mqtt_recv_thread(void *args)
{
	int32_t res = STATE_SUCCESS;

	while (g_mqtt_recv_thread_running) {
		res = aiot_mqtt_recv(args);
		if (res < STATE_SUCCESS) {
			if (res == STATE_USER_INPUT_EXEC_DISABLED) {
				break;
			}
			sleep(1);
		}
	}
	return NULL;
}

void *mqtt_start(const char *pkey, const char *dname, const char *dsec, const char *host)
{
	int32_t res = STATE_SUCCESS;
	void *mqtt_handle = NULL;
	aiot_sysdep_network_cred_t cred; /* 安全凭据结构体, 如果要用TLS, 这个结构体中配置CA证书等参数 */

	/* 配置SDK的底层依赖 */
	aiot_sysdep_set_portfile(&g_aiot_sysdep_portfile);
	/* 配置SDK的日志输出 */
	aiot_state_set_logcb(demo_state_logcb);
	const char *product_key = pkey;
	const char *device_name = dname;
	const char *device_secret = dsec;
	const char *mqtt_host = host;
	/* 创建SDK的安全凭据, 用于建立TLS连接 */
	memset(&cred, 0, sizeof(aiot_sysdep_network_cred_t));
	cred.option = AIOT_SYSDEP_NETWORK_CRED_SVRCERT_CA; /* 使用RSA证书校验MQTT服务端 */
	cred.max_tls_fragment = 16384; /* 最大的分片长度为16K, 其它可选值还有4K, 2K, 1K, 0.5K */
	cred.sni_enabled = 1; /* TLS建连时, 支持Server Name Indicator */
	cred.x509_server_cert = ali_ca_cert; /* 用来验证MQTT服务端的RSA根证书 */
	cred.x509_server_cert_len = strlen(ali_ca_cert); /* 用来验证MQTT服务端的RSA根证书长度 */

	/* 创建1个MQTT客户端实例并内部初始化默认参数 */
	mqtt_handle = aiot_mqtt_init();
	if (mqtt_handle == NULL) {
		printf("aiot_mqtt_init failed\n");
		return NULL;
	}

	/* TODO: 如果以下代码不被注释, 则例程会用TCP而不是TLS连接云平台 */
	/*
    {
        memset(&cred, 0, sizeof(aiot_sysdep_network_cred_t));
        cred.option = AIOT_SYSDEP_NETWORK_CRED_NONE;
    }
    */

	/* 配置MQTT服务器地址 */
	aiot_mqtt_setopt(mqtt_handle, AIOT_MQTTOPT_HOST, (void *)mqtt_host);
	/* 配置MQTT服务器端口 */
	aiot_mqtt_setopt(mqtt_handle, AIOT_MQTTOPT_PORT, (void *)&port);
	/* 配置设备productKey */
	aiot_mqtt_setopt(mqtt_handle, AIOT_MQTTOPT_PRODUCT_KEY, (void *)product_key);
	/* 配置设备deviceName */
	aiot_mqtt_setopt(mqtt_handle, AIOT_MQTTOPT_DEVICE_NAME, (void *)device_name);
	/* 配置设备deviceSecret */
	aiot_mqtt_setopt(mqtt_handle, AIOT_MQTTOPT_DEVICE_SECRET, (void *)device_secret);
	/* 配置网络连接的安全凭据, 上面已经创建好了 */
	aiot_mqtt_setopt(mqtt_handle, AIOT_MQTTOPT_NETWORK_CRED, (void *)&cred);
	/* 配置MQTT默认消息接收回调函数 */
	aiot_mqtt_setopt(mqtt_handle, AIOT_MQTTOPT_RECV_HANDLER, (void *)demo_mqtt_default_recv_handler);
	/* 配置MQTT事件回调函数 */
	aiot_mqtt_setopt(mqtt_handle, AIOT_MQTTOPT_EVENT_HANDLER, (void *)demo_mqtt_event_handler);

	/* 与服务器建立MQTT连接 */
	res = aiot_mqtt_connect(mqtt_handle);
	if (res < STATE_SUCCESS) {
		/* 尝试建立连接失败, 销毁MQTT实例, 回收资源 */
		aiot_mqtt_deinit(&mqtt_handle);
		printf("aiot_mqtt_connect failed: -0x%04X\n\r\n", -res);
		printf("please check variables like mqtt_host, produt_key, device_name, device_secret in demo\r\n");
		return NULL;
	}

	char topic[256];
	snprintf(topic, sizeof(topic), "/%s/%s/user/scene/issue", g_conf->self["ProductionKey"].c_str(), g_conf->self["DevName"].c_str());
	if (aiot_mqtt_sub(mqtt_handle, topic, NULL, 1, NULL) < STATE_SUCCESS) {
		perror("aiot_mqtt_sub");
		exit(0);
	}

	snprintf(topic, sizeof(topic), "/%s/%s/user/get", g_conf->self["ProductionKey"].c_str(), g_conf->self["DevName"].c_str());
	if (aiot_mqtt_sub(mqtt_handle, topic, NULL, 1, NULL) < STATE_SUCCESS) {
		perror("aiot_mqtt_sub");
		exit(0);
	}

	snprintf(topic, sizeof(topic), "/%s/%s/user/issue", g_conf->self["ProductionKey"].c_str(), g_conf->self["DevName"].c_str());
	if (aiot_mqtt_sub(mqtt_handle, topic, NULL, 1, NULL) < STATE_SUCCESS) {
		perror("aiot_mqtt_sub");
		exit(0);
	}

	/* 创建一个单独的线程, 专用于执行aiot_mqtt_process, 它会自动发送心跳保活, 以及重发QoS1的未应答报文 */
	g_mqtt_process_thread_running = 1;
	res = pthread_create(&g_mqtt_process_thread, NULL, demo_mqtt_process_thread, mqtt_handle);
	if (res < 0) {
		printf("pthread_create demo_mqtt_process_thread failed: %d\n", res);
		return NULL;
	}

	/* 创建一个单独的线程用于执行aiot_mqtt_recv, 它会循环收取服务器下发的MQTT消息, 并在断线时自动重连 */
	g_mqtt_recv_thread_running = 1;
	res = pthread_create(&g_mqtt_recv_thread, NULL, demo_mqtt_recv_thread, mqtt_handle);
	if (res < 0) {
		printf("pthread_create demo_mqtt_recv_thread failed: %d\n", res);
		return NULL;
	}

	return mqtt_handle;
}

int32_t demo_mqtt_stop(void **handle)
{
	int32_t res = STATE_SUCCESS;
	void *mqtt_handle = NULL;

	mqtt_handle = *handle;

	g_mqtt_process_thread_running = 0;
	g_mqtt_recv_thread_running = 0;
	pthread_join(g_mqtt_process_thread, NULL);
	pthread_join(g_mqtt_recv_thread, NULL);

	/* 断开MQTT连接 */
	res = aiot_mqtt_disconnect(mqtt_handle);
	if (res < STATE_SUCCESS) {
		aiot_mqtt_deinit(&mqtt_handle);
		printf("aiot_mqtt_disconnect failed: -0x%04X\n", -res);
		return -1;
	}

	/* 销毁MQTT实例 */
	res = aiot_mqtt_deinit(&mqtt_handle);
	if (res < STATE_SUCCESS) {
		printf("aiot_mqtt_deinit failed: -0x%04X\n", -res);
		return -1;
	}

	return 0;
}

void demo_subdev_recv_handler(void *handle, const aiot_subdev_recv_t *packet, void *user_data)
{
	nlohmann::json js;
	switch (packet->type) {
	case AIOT_SUBDEVRECV_TOPO_ADD_REPLY:
		js["Reply"] = "AIOT_SUBDEVRECV_TOPO_ADD_REPLY";
		break;
	case AIOT_SUBDEVRECV_TOPO_DELETE_REPLY:
		js["Reply"] = "AIOT_SUBDEVRECV_TOPO_DELETE_REPLY";
		break;
	case AIOT_SUBDEVRECV_TOPO_GET_REPLY:
		js["Reply"] = "AIOT_SUBDEVRECV_TOPO_GET_REPLY";
		break;
	case AIOT_SUBDEVRECV_BATCH_LOGIN_REPLY:
		js["Reply"] = "AIOT_SUBDEVRECV_BATCH_LOGIN_REPLY";
		break;
	case AIOT_SUBDEVRECV_BATCH_LOGOUT_REPLY:
		js["Reply"] = "AIOT_SUBDEVRECV_BATCH_LOGOUT_REPLY";
		break;
	case AIOT_SUBDEVRECV_SUB_REGISTER_REPLY:
		js["Reply"] = "AIOT_SUBDEVRECV_SUB_REGISTER_REPLY";
		break;
	case AIOT_SUBDEVRECV_PRODUCT_REGISTER_REPLY:
		js["Reply"] = "AIOT_SUBDEVRECV_PRODUCT_REGISTER_REPLY";
		break;
	case AIOT_SUBDEVRECV_TOPO_CHANGE_NOTIFY:
		js["Reply"] = "AIOT_SUBDEVRECV_TOPO_CHANGE_NOTIFY";
		break;
	default:
		return;
	}

	js["Params"] = {

		{ "code", packet->data.generic_reply.code },

		{ "data", json::parse(packet->data.generic_reply.data) },

		{ "message", packet->data.generic_reply.message == NULL ? "NULL" : packet->data.generic_reply.message }

	};
	string tok = js.dump(4);
	send_msg(CTRL_QUEUE, tok.c_str(), tok.size());

	if (packet->type == AIOT_SUBDEVRECV_TOPO_CHANGE_NOTIFY) {
		printf("msgid        : %d\n", packet->data.generic_notify.msg_id);
		printf("product key  : %s\n", packet->data.generic_notify.product_key);
		printf("device name  : %s\n", packet->data.generic_notify.device_name);
		printf("params       : %s\n", packet->data.generic_notify.params);
	} else {
		printf("msgid        : %d\n", packet->data.generic_reply.msg_id);
		printf("code         : %d\n", packet->data.generic_reply.code);
		printf("product key  : %s\n", packet->data.generic_reply.product_key);
		printf("device name  : %s\n", packet->data.generic_reply.device_name);
		printf("message      : %s\n", (packet->data.generic_reply.message == NULL) ? ("NULL") : (packet->data.generic_reply.message));
		printf("data         : %s\n", packet->data.generic_reply.data);
	}
}

void *subdev_proc_init(void *mqtt_handle)
{
	void *subdev_handle = aiot_subdev_init();
	if (subdev_handle == NULL) {
		printf("aiot_subdev_init failed\n");
		demo_mqtt_stop(&mqtt_handle);
		return NULL;
	}

	aiot_subdev_setopt(subdev_handle, AIOT_SUBDEVOPT_MQTT_HANDLE, mqtt_handle);
	aiot_subdev_setopt(subdev_handle, AIOT_SUBDEVOPT_RECV_HANDLER, (void *)demo_subdev_recv_handler);
	return subdev_handle;
}

/***********************************OTA PROCE*************************************************/
/* 下载收包回调, 用户调用 aiot_download_recv() 后, SDK收到数据会进入这个函数, 把下载到的数据交给用户 */
/* TODO: 一般来说, 设备升级时, 会在这个回调中, 把下载到的数据写到Flash上 */
pthread_t g_download_thread = 0;
struct DL_DATA {
	int last_percent;
	int dl_fd;
} dl_data;
void demo_download_recv_handler(void *handle, const aiot_download_recv_t *packet, void *userdata)
{
	uint32_t data_buffer_len = 0;
	uint32_t last_percent = 0;
	int32_t percent = 0;
	DL_DATA *dl_usr = (DL_DATA *)userdata;

	/* 目前只支持 packet->type 为 AIOT_DLRECV_HTTPBODY 的情况 */
	if (!packet || AIOT_DLRECV_HTTPBODY != packet->type) {
		return;
	}
	percent = packet->data.percent;

	/* userdata可以存放 demo_download_recv_handler() 的不同次进入之间, 需要共享的数据 */
	/* 这里用来存放上一次进入本回调函数时, 下载的固件进度百分比 */
	if (userdata) {
		last_percent = dl_usr->last_percent;
	}

	data_buffer_len = packet->data.len;

	/* 如果 percent 为负数, 说明发生了收包异常或者digest校验错误 */
	if (percent < 0) {
		printf("exception: percent = %d\r\n", percent);
		if (userdata) {
			free(userdata);
		}
		return;
	}

	/*
     * TODO: 下载一段固件成功, 这个时候, 用户应该将
     *       起始地址为 packet->data.buffer, 长度为 packet->data.len 的内存, 保存到flash上
     *
     *       如果烧写flash失败, 还应该调用 aiot_download_report_progress(handle, -4) 将失败上报给云平台
     *       备注:协议中, 与云平台商定的错误码在 aiot_ota_protocol_errcode_t 类型中, 例如
     *           -1: 表示升级失败
     *           -2: 表示下载失败
     *           -3: 表示校验失败
     *           -4: 表示烧写失败
     *
     *       详情可见 https://help.aliyun.com/document_detail/85700.html
     */
	if (write(dl_usr->dl_fd, packet->data.buffer, packet->data.len) < 0) {
		aiot_download_report_progress(handle, -2);
	}

	/* percent 入参的值为 100 时, 说明SDK已经下载固件内容全部完成 */
	if (percent == 100) {
		/*
         * TODO: 这个时候, 一般用户就应该完成所有的固件烧录, 保存当前工作, 重启设备, 切换到新的固件上启动了
                 并且, 新的固件必须要以

                 aiot_ota_report_version(ota_handle, new_version);

                 这样的操作, 将升级后的新版本号(比如1.0.0升到1.1.0, 则new_version的值是"1.1.0")上报给云平台
                 云平台收到了新的版本号上报后, 才会判定升级成功, 否则会认为本次升级失败了
                 如果下载成功后升级失败, 还应该调用 aiot_download_report_progress(handle, -1) 将失败上报给云平台
         */
		close(dl_usr->dl_fd);
	}

	/* 简化输出, 只有距离上次的下载进度增加5%以上时, 才会打印进度, 并向服务器上报进度 */
	if (percent - last_percent >= 5 || percent == 100) {
		printf("download %03d%% done, +%d bytes\r\n", percent, data_buffer_len);
		aiot_download_report_progress(handle, percent);

		if (userdata) {
			dl_usr->last_percent = percent;
		}
	}
}

/* 执行aiot_download_recv的线程, 实现固件内容的请求和接收 */
void *demo_ota_download_thread(void *dl_handle)
{
	int32_t ret = 0;

	printf("starting download thread in 2 seconds ......\r\n");
	sleep(2);

	/* 向固件服务器请求下载 */
	/*
     * TODO: 下面这样的写法, 就是以1个请求, 获取全部的固件内容
     *       设备资源比较少, 或者网络较差时, 也可以分段下载, 需要组合
     *
     *       aiot_download_setopt(dl_handle, AIOT_DLOPT_RANGE_START, ...);
     *       aiot_download_setopt(dl_handle, AIOT_DLOPT_RANGE_END, ...);
     *       aiot_download_send_request(dl_handle);
     *
     *       实现, 这种情况下, 需要把以上组合语句放置到循环中, 多次 send_request 和 recv
     *
     */
	aiot_download_send_request(dl_handle);
	while (1) {
		/* 从网络收取服务器回应的固件内容 */
		ret = aiot_download_recv(dl_handle);

		/* 固件全部下载完时, aiot_download_recv() 的返回值会等于 STATE_DOWNLOAD_FINISHED, 否则是当次获取的字节数 */
		if (STATE_DOWNLOAD_FINISHED == ret) {
			printf("download completed\r\n");
			break;
		}
		if (STATE_DOWNLOAD_RENEWAL_REQUEST_SENT == ret) {
			printf("download renewal request has been sent successfully\r\n");
			continue;
		}
		if (ret <= STATE_SUCCESS) {
			printf("download failed, error code is %d, try to send renewal request\r\n", ret);
			continue;
		}
	}

	/* 下载所有固件内容完成, 销毁下载会话, 线程自行退出 */
	aiot_download_deinit(&dl_handle);
	json js;
	js["Active"] = "UPGRADE";
	string tok = js.dump(4);
	send_msg(SERVICE_QUEUE_REQ, tok.c_str(), tok.size());

	printf("download thread exit\r\n");

	return NULL;
}

/* 用户通过 aiot_ota_setopt() 注册的OTA消息处理回调, 如果SDK收到了OTA相关的MQTT消息, 会自动识别, 调用这个回调函数 */
void demo_ota_recv_handler(void *ota_handle, aiot_ota_recv_t *ota_msg, void *userdata)
{
	switch (ota_msg->type) {
	case AIOT_OTARECV_FOTA: {
		uint32_t res = 0;
		uint16_t port = 443;
		uint32_t max_buffer_len = (8 * 1024);
		aiot_sysdep_network_cred_t cred;
		void *dl_handle = NULL;
		void *last_percent = NULL;

		if (NULL == ota_msg->task_desc || ota_msg->task_desc->protocol_type != AIOT_OTA_PROTOCOL_HTTPS) {
			break;
		}

		dl_handle = aiot_download_init();
		if (NULL == dl_handle) {
			break;
		}
		printf("OTA target firmware version: %s, size: %u Bytes \r\n", ota_msg->task_desc->version, ota_msg->task_desc->size_total);

		if (NULL != ota_msg->task_desc->extra_data) {
			printf("extra data: %s\r\n", ota_msg->task_desc->extra_data);
		}

		memset(&cred, 0, sizeof(aiot_sysdep_network_cred_t));
		cred.option = AIOT_SYSDEP_NETWORK_CRED_SVRCERT_CA;
		cred.max_tls_fragment = 16384;
		cred.x509_server_cert = ali_ca_cert;
		cred.x509_server_cert_len = strlen(ali_ca_cert);

		/* 设置下载时为TLS下载 */
		aiot_download_setopt(dl_handle, AIOT_DLOPT_NETWORK_CRED, (void *)(&cred));
		/* 设置下载时访问的服务器端口号 */
		aiot_download_setopt(dl_handle, AIOT_DLOPT_NETWORK_PORT, (void *)(&port));
		/* 设置下载的任务信息, 通过输入参数 ota_msg 中的 task_desc 成员得到, 内含下载地址, 固件大小, 固件签名等 */
		aiot_download_setopt(dl_handle, AIOT_DLOPT_TASK_DESC, (void *)(ota_msg->task_desc));
		/* 设置下载内容到达时, SDK将调用的回调函数 */
		aiot_download_setopt(dl_handle, AIOT_DLOPT_RECV_HANDLER, (void *)(demo_download_recv_handler));
		/* 设置单次下载最大的buffer长度, 每当这个长度的内存读满了后会通知用户 */
		aiot_download_setopt(dl_handle, AIOT_DLOPT_BODY_BUFFER_MAX_LEN, (void *)(&max_buffer_len));
		/* 设置 AIOT_DLOPT_RECV_HANDLER 的不同次调用之间共享的数据, 比如例程把进度存在这里 */
		// last_percent = malloc(sizeof(uint32_t));
		// if (NULL == last_percent) {
		// 	aiot_download_deinit(&dl_handle);
		// 	break;
		// }
		// memset(last_percent, 0, sizeof(uint32_t));
		// aiot_download_setopt(dl_handle, AIOT_DLOPT_USERDATA, (void *)last_percent);
		dl_data.last_percent = 0;
		dl_data.dl_fd = open("/tmp/sys_upgrade.bin", O_RDWR | O_CREAT | O_TRUNC, 0666);
		if (dl_data.dl_fd < 0) {
			return;
		}
		aiot_download_setopt(dl_handle, AIOT_DLOPT_USERDATA, (void *)&dl_data);

		/* 启动专用的下载线程, 去完成固件内容的下载 */
		res = pthread_create(&g_download_thread, NULL, demo_ota_download_thread, dl_handle);
		if (res != 0) {
			printf("pthread_create demo_ota_download_thread failed: %d\r\n", res);
			aiot_download_deinit(&dl_handle);
		} else {
			/* 下载线程被设置为 detach 类型, 固件内容获取完毕后可自主退出 */
			pthread_detach(g_download_thread);
		}
		break;
	}

	default:
		break;
	}
}

void *ota_proc_init(void *mqtt_handle, const char *cur_version)
{
	int32_t res = STATE_SUCCESS;
	void *ota_handle = aiot_ota_init();
	if (NULL == ota_handle) {
		printf("aiot_ota_init failed\r\n");
		aiot_mqtt_deinit(&mqtt_handle);
		return NULL;
	}
	/* 用以下语句, 把OTA会话和MQTT会话关联起来 */
	aiot_ota_setopt(ota_handle, AIOT_OTAOPT_MQTT_HANDLE, mqtt_handle);
	/* 用以下语句, 设置OTA会话的数据接收回调, SDK收到OTA相关推送时, 会进入这个回调函数 */
	aiot_ota_setopt(ota_handle, AIOT_OTAOPT_RECV_HANDLER, (void *)demo_ota_recv_handler);
	res = aiot_ota_report_version(ota_handle, (char *)cur_version);
	if (res < STATE_SUCCESS) {
		printf("aiot_ota_report_version failed: -0x%04X\r\n", -res);
		exit(0);
	}
	return ota_handle;
}

/********************RA_PROC************/
static pthread_t g_ra_process_thread;
aiot_ra_service_t services[] = {
	{
		.type = "_SSH",
		.ip = "127.0.0.1",
		.port = 22,
	},
};

void ra_event_cb(void *handle, const aiot_ra_event_t *event, void *userdata)
{
	switch (event->type) {
	case AIOT_RA_EVT_CONNECT:
		printf("ra_event_cb AIOT_RA_EVT_CONNECT %s \r\n", event->tunnel_id);
		/* TODO: 告知websocket建连成功, 不可以在这里调用耗时较长的阻塞函数 */
		break;
	case AIOT_RA_EVT_DISCONNECT:
		printf("ra_event_cb AIOT_RA_EVT_DISCONNECT %s \r\n", event->tunnel_id);
		/* TODO: 告知websocket掉线, 不可以在这里调用耗时较长的阻塞函数 */
		break;
	case AIOT_RA_EVT_OPEN_WEBSOCKET:
		printf("ra_event_cb AIOT_RA_EVT_OPEN_WEBSOCKET %s \r\n", event->tunnel_id);
		/* TODO: 告知RA接收到打开websocket链接命令, 不可以在这里调用耗时较长的阻塞函数 */
		break;
	case AIOT_RA_EVT_CLOSE_WEBSOCKET:
		printf("ra_event_cb AIOT_RA_EVT_CLOSE_WEBSOCKET %s \r\n", event->tunnel_id);
		/* TODO: 告知RA接收到关闭websocket链接命令, 不可以在这里调用耗时较长的阻塞函数 */
		break;
	}
}

void *ra_proc_init(void *mqtt_handle)
{
	/* 创建1个RA实例并内部初始化默认参数 */
	void *ra_handle = aiot_ra_init();
	if (ra_handle == NULL) {
		printf("aiot_ra_init failed\n");
		return NULL;
	}

	aiot_sysdep_network_cred_t cred; /* 安全凭据结构体, 如果要用TLS, 这个结构体中配置CA证书等参数 */
	memset(&cred, 0, sizeof(aiot_sysdep_network_cred_t));
	cred.option = AIOT_SYSDEP_NETWORK_CRED_SVRCERT_CA; /* 使用RSA证书校验MQTT服务端 */
	cred.max_tls_fragment = 16384; /* 最大的分片长度为16K, 其它可选值还有4K, 2K, 1K, 0.5K */
	cred.sni_enabled = 1; /* TLS建连时, 支持Server Name Indicator */
	cred.x509_server_cert = ali_ca_cert; /* 用来验证MQTT服务端的RSA根证书 */
	cred.x509_server_cert_len = strlen(ali_ca_cert); /* 用来验证MQTT服务端的RSA根证书长度 */

	/* 配置MQTT句柄，ra内部会订阅MQTT的消息 */
	aiot_ra_setopt(ra_handle, AIOT_RAOPT_MQTT_HANDLE, mqtt_handle);
	/* 配置网络连接的安全凭据, 上面已经创建好了 */
	aiot_ra_setopt(ra_handle, AIOT_RAOPT_NETWORK_CRED, (void *)&cred);
	/* 配置RA内部事件回调函数， 可选*/
	aiot_ra_setopt(ra_handle, AIOT_RAOPT_EVENT_HANDLER, (void *)ra_event_cb);
	/* 配置本地可支持的服务 */
	for (int i = 0; i < sizeof(services) / sizeof(aiot_ra_service_t); i++) {
		aiot_ra_setopt(ra_handle, AIOT_RAOPT_ADD_SERVICE, (void *)&services[i]);
	}
	/*开启线程，运行RA服务*/
	pthread_attr_t attr;
	pthread_attr_init(&attr);
	if (0 != pthread_create(&g_ra_process_thread, &attr, aiot_ra_start, (void *)ra_handle)) {
		printf("creat remote_proxy_thread error!");
		return NULL;
	}
	return ra_handle;
}
